package net.bwie.zg6.spark

import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

/**
 *京东用户日志离线分析
 *
 * @author xuanyu
 * @date 2025/6/24
 */
object SparkJdLogSqlApp {

    def main(args: Array[String]): Unit = {
        // 1. SparkSession会话对象
        val spark: SparkSession = SparkSession.builder()
            .appName("SparkJdLogSqlApp")
            .master("local[2]")
            .config("spark.sql.shuffle.partitions", "2")
            .getOrCreate()
        import spark.implicits._

        spark.sparkContext.setLogLevel("WARN")

        // 2. 读取数据
        val logDF: DataFrame = spark.read
            .option("sep", ",")
            .schema("ip STRING, user_id STRING, url STRING, access_time STRING")
            .csv("zg6-spark-lecture/src/main/resources/jd-behavior.log")
//        logDF.printSchema()
//        logDF.show(10, truncate = false)

        // 注册临时表
        logDF.createTempView("tmp_jd_log")
        // spark.sql("SELECT * FROM tmp_jd_log").show(false)

        // 4. 分析1：京东每日的UV，降序排序，打印输出；
        println("4. 分析1：京东每日的UV，降序排序，打印输出；")
        val df4 = spark.sql(
            """
              |SELECT
              | substring(access_time, 1, 10) AS day_str
              | , count(distinct user_id) AS uv
              |FROM tmp_jd_log
              |GROUP BY substring(access_time, 1, 10)
              |""".stripMargin
        )
        df4.show(false)

        // 5. 分析2：京东每日浏览量PV，降序排序，打印输出；
        println("5. 分析2：京东每日浏览量PV，降序排序，打印输出；")
        val df5 = spark.sql(
            """
              |SELECT
              | substring(access_time, 1, 10) AS day_str
              | , count(url) AS pv
              |FROM tmp_jd_log
              |GROUP BY substring(access_time, 1, 10)
              |""".stripMargin
        )
        df5.show(false)


        // 6. 分析3：京东每个url每日访客数UV，降序排序，打印输出；
        println("6. 分析3：京东每个url每日访客数UV，降序排序，打印输出；")
        val df6 = spark.sql(
            """
              |SELECT
              | url
              | , substring(access_time, 1, 10) AS day_str
              | , count(distinct user_id) AS uv
              |FROM tmp_jd_log
              |GROUP BY url, substring(access_time, 1, 10)
              |""".stripMargin
        )
        df6.show(false)


        // 7. 分析4：京东每个url每日浏览量PV，降序排序，保存MySQL表：result_pv；
        println("7. 分析4：京东每个url每日浏览量PV，降序排序，保存MySQL表：result_pv；")
        val df7 = spark.sql(
            """
              |SELECT
              | url
              | , substring(access_time, 1, 10) AS day_str
              | , count(url) AS pv
              |FROM tmp_jd_log
              |GROUP BY url, substring(access_time, 1, 10)
              |ORDER BY pv DESC
              |""".stripMargin
        )
        df7.show(false)

        df7.write
            .mode(SaveMode.Overwrite)
            .format("jdbc")
            .option("url", "jdbc:mysql://node101:3306/db_test?createDatabaseIfNotExist=true&useSSL=false&useUnicode=true&characterEncoding=UTF-8")
            .option("dbtable", "db_test.result_pv")
            .option("user", "root")
            .option("password", "123456")
            .save()
        // todo 实际项目中，如果将结果保存MySQL数据库表中，采用原生JDBC方式书写，实现主键存在更新，不存在插入。

        // 8. SparkSQL定义UDF函数：从字段datetime获取每天日期；
        spark.udf.register(
            "extract_date", // 函数名称
            (accessTime: String) => {
                accessTime.substring(0, 10)
            }
        )

        // 9. 分析5：编写SQL语句，使用UDF函数，获取日期，打印输出；
        println("9. 分析5：编写SQL语句，使用UDF函数，获取日期，打印输出；")
        val df9: DataFrame = spark.sql(
            """
              |SELECT
              | access_time
              | , substring(access_time, 1, 10) AS day_str
              | , extract_date(access_time) AS date_str
              |FROM tmp_jd_log
              |""".stripMargin
        )
        df9.show(false)


        println("=====================================================================")
        // 自定义udf函数第2种方式
        import org.apache.spark.sql.functions.udf
        val transform_date: UserDefinedFunction = udf(
            (accessTime: String) => {
                accessTime.substring(0, 10)
            }
        )
        logDF.select(transform_date($"access_time").as("date_str")).show(false)


        // 关闭
        spark.stop()
    }

}
